package cn.lokei.gateway.handler.mqtt;

import java.io.IOException;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import cn.lokei.gateway.entity.IotChannel;
import cn.lokei.gateway.handler.message.ReceiveHandler;
import cn.lokei.gateway.service.ChannelSupervise;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttSubscribePayload;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;

public class MqttHandler extends ChannelInboundHandlerAdapter {

	private Logger log = LoggerFactory.getLogger(this.getClass());

	private ChannelSupervise channelSupervise = ChannelSupervise.getBean(ChannelSupervise.class);

	private RestTemplate restTemplate = new RestTemplate();

	private ReceiveHandler receiveHandler = new ReceiveHandler();

	// @Autowired
	// KafkaProducer kafkaProducer;

	/**
	 * 客户端与服务端第一次建立连接时执行 在channelActive方法之前执行
	 */
	@Override
	public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
		super.channelRegistered(ctx);
		channelSupervise.addChannel(ctx.channel());
	}

	/**
	 * 客户端与服务端 断连时执行 channelInactive方法之后执行
	 */
	@Override
	public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
		super.channelUnregistered(ctx);
		channelSupervise.removeChannel(ctx.channel());
	}

	/**
	 * 从客户端收到新的数据时，这个方法会在收到消息时被调用
	 */
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException {
		MqttMessage mqttMessage = (MqttMessage) msg;
		
		MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
		Channel channel = ctx.channel();

		if(!mqttFixedHeader.messageType().equals(MqttMessageType.PINGREQ)) {
			log.info("info--" + mqttMessage.toString());
		}

		if (mqttFixedHeader.messageType().equals(MqttMessageType.CONNECT)) {
			// 在一个网络连接上，客户端只能发送一次CONNECT报文。服务端必须将客户端发送的第二个CONNECT报文当作协议违规处理并断开客户端的连接
			// to do 建议connect消息单独处理，用来对客户端进行认证管理等 这里直接返回一个CONNACK消息
			MqttConnectPayload payload = (MqttConnectPayload) mqttMessage.payload();
			IotChannel iotChannel = channelSupervise.findChannel(channel.id().asLongText());
			if (iotChannel != null) {
				iotChannel.set_connected(true);
				iotChannel.setClient_id(payload.clientIdentifier());
				iotChannel.setUsername(payload.userName());
				iotChannel.setPassword(payload.passwordInBytes());
				try {
					// RestTemplate restTemplate = RestTemplate.getBean("restTemplate");
					// kafkaProducer.send(mqttMessage);
					ResponseEntity<String> responseEntity = restTemplate.getForEntity(
							"https://iot.lokei.cn/auth/device",
							String.class);
					HttpStatus statusCode = responseEntity.getStatusCode(); // 获取响应码
					if (statusCode == HttpStatus.OK) {
						iotChannel.set_registered(true);
					} else {
						iotChannel.set_registered(true);
					}
				} catch (Exception e) {
					iotChannel.set_registered(false);
				}
			}
			MqttMsgBack.connack(channel, mqttMessage);
		}

		switch (mqttFixedHeader.messageType()) {
			case PUBLISH: // 客户端发布消息
				// PUBACK报文是对QoS 1等级的PUBLISH报文的响应
				// System.out.println("123");
				// KafkaProducer kafkaProducer = KafkaProducer.getBean("kafkaProducer");
				// kafkaProducer.send(mqttMessage);
				receiveHandler.receiveMessage(mqttMessage);
				MqttMsgBack.puback(channel, mqttMessage);
				break;
			case PUBREL: // 发布释放
				// PUBREL报文是对PUBREC报文的响应
				// to do
				MqttMsgBack.pubcomp(channel, mqttMessage);
				break;
			case SUBSCRIBE: // 客户端订阅主题
				// 客户端向服务端发送SUBSCRIBE报文用于创建一个或多个订阅，每个订阅注册客户端关心的一个或多个主题。
				// 为了将应用消息转发给与那些订阅匹配的主题，服务端发送PUBLISH报文给客户端。
				// SUBSCRIBE报文也（为每个订阅）指定了最大的QoS等级，服务端根据这个发送应用消息给客户端
				// to do
				MqttSubscribePayload payload = (MqttSubscribePayload) mqttMessage.payload();
				List<MqttTopicSubscription> topicSubscriptions = payload.topicSubscriptions();
				for (MqttTopicSubscription mqttTopicSubscription : topicSubscriptions) {
					channelSupervise.subTopic(mqttTopicSubscription.topicName(), channel);
				}
				MqttMsgBack.suback(channel, mqttMessage);
				break;
			case UNSUBSCRIBE: // 客户端取消订阅
				// 客户端发送UNSUBSCRIBE报文给服务端，用于取消订阅主题
				// to do
				MqttUnsubscribeMessage mqttUnsubscribeMessage = (MqttUnsubscribeMessage) mqttMessage;
				channelSupervise.unsubTopic(mqttUnsubscribeMessage.payload().topics(), channel);
				MqttMsgBack.unsuback(channel, mqttMessage);
				break;
			case PINGREQ: // 客户端发起心跳
				// 客户端发送PINGREQ报文给服务端的
				// 在没有任何其它控制报文从客户端发给服务的时，告知服务端客户端还活着
				// 请求服务端发送 响应确认它还活着，使用网络以确认网络连接没有断开
				MqttMsgBack.pingresp(channel, mqttMessage);
				break;
			case DISCONNECT: // 客户端主动断开连接
				// DISCONNECT报文是客户端发给服务端的最后一个控制报文， 服务端必须验证所有的保留位都被设置为0
				// to do
				break;
			default:
				break;
		}
	}

	/**
	 * 从客户端收到新的数据、读取完成时调用
	 */
	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
	}

	/**
	 * 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
	 */
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		super.exceptionCaught(ctx, cause);
		ctx.close();
	}

	/**
	 * 客户端与服务端第一次建立连接时执行
	 */
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		super.channelActive(ctx);
	}

	/**
	 * 客户端与服务端 断连时执行
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
		super.channelInactive(ctx);
	}

	/**
	 * 服务端 当读超时时 会调用这个方法
	 */
	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException {
		super.userEventTriggered(ctx, evt);
		ctx.close();
	}

	@Override
	public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
		super.channelWritabilityChanged(ctx);
	}
}
